package com.hrz.datasync.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hrz.datasync.pojo.primary.TpAdZone;
import com.hrz.datasync.pojo.secondary.BbqAdZone;
import com.hrz.datasync.service.BbqAdZoneService;
import com.hrz.datasync.service.TpAdZoneService;
import lombok.extern.java.Log;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @作 者: 侯建军
 * @日 期: 2018/11/05 15:45:14
 * @描 述:
 */
@Component
@Log
public class KafkaConsumer {
    @Autowired
    private TpAdZoneService tpAdZoneService;
    @Autowired
    private BbqAdZoneService bbqAdZoneService;
    /**
     * 监听kafka.tut 的 topic
     *
     * @param record
     * @param topic  topic
     */
    @KafkaListener(id = "tpzdzone", topics = {"tpzdzone","bbqzdzone"})
    public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        //判断是否NULL
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if (kafkaMessage.isPresent()) {
            //获取消息
            Object message = kafkaMessage.get();
            log.info("接受->topic:" + topic);

            //写数据到数据库
            switch (topic) {
                case "bbqzdzone":
                    log.info("bbqzdzone->接受->Record:" + record);
                    log.info("接受消息:" + message);
                    //
                    BbqAdZone bbqAdZone = JSON.parseObject(message.toString(),BbqAdZone.class);
                    tpAdZoneService.addTpAdZone(bbqAdZone);
                    break;
                case "tpzdzone":
                    log.info("接受->Record:" + record);
                    log.info("接受消息:" + message);

                    TpAdZone tpAdZone = JSON.parseObject(message.toString(),TpAdZone.class);
                    log.info(tpAdZone.toString());
                    log.info(bbqAdZoneService.toString());
                    bbqAdZoneService.addBbqAdZone(tpAdZone);
                    break;
                default:
                    break;
            }
        }
    }

}
